package cn.t.modules.scheduler.core.schedule;

import cn.t.modules.scheduler.common.enums.ScheduleStatus;
import cn.t.modules.scheduler.common.enums.ScheduleType;
import cn.t.modules.scheduler.common.utils.TimeUtil;
import cn.t.modules.scheduler.core.SqlProcessor;
import cn.t.modules.scheduler.core.schedule.repository.po.SchedulePo;
import cn.t.modules.timewheel.core.hierarchical.AbstractTask;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;

import javax.annotation.Resource;
import java.util.stream.Collectors;

/**
 * @author 陶敏麒
 * @date 2024/3/6 15:36
 */
@Component
@Slf4j
public class Scheduler {

    @Resource
    private SqlProcessor sqliteProcessor;

    /**
     * 任务执行队列
     */
    @Resource
    private ScheduleQueue scheduleQueue;

    @Transactional(rollbackFor = Exception.class)
    public String addTask(AbstractTask task) {
        var po = new SchedulePo();
        po.setScheduleId(task.getScheduleId());
        po.setContent(task.getContent());
        po.setScheduleName(task.getScheduleName());
        po.setExecTimeTs(task.getExecTs());
        po.setVersion(0);
        po.setType(ScheduleType.valueOf(task.type()));
        var date = TimeUtil.currentTime();
        po.setCreateTime(date);
        po.setUpdateTime(date);
        // 判断时间(1分钟之内的进入执行状态)
        if (System.currentTimeMillis() + 60000 >= task.getExecTs()) {
            // 直接执行
            po.setStatus(ScheduleStatus.RUNNING);
        } else {
            po.setStatus(ScheduleStatus.ACCEPTED);
        }
        // 入库
        this.sqliteProcessor.saveSchedule(po);
        task.setTaskId(po.getId());
        if (ScheduleStatus.RUNNING.equals(po.getStatus())) {
            // 直接入队
            this.scheduleQueue.offerSchedule(po);
        }
        return po.getScheduleId();
    }

    /**
     * 60秒扫一次,单线程扫描
     */
    @Scheduled(fixedDelay = 60000, initialDelay = 10000)
    protected void scan() {
        var list = this.sqliteProcessor.queryAcceptedScheduleList(this.scheduleQueue.remainingCapacity());
        if (CollectionUtils.isEmpty(list)) {
            return;
        }
        // 批量更新状态
        this.sqliteProcessor.updateStatusBatch(list.stream().map(SchedulePo::getId).collect(Collectors.toSet()), ScheduleStatus.RUNNING, "");
        for (SchedulePo schedulePo : list) {
            try {
                // 入队(入队失败会重置状态)
                this.scheduleQueue.offerSchedule(schedulePo);
            } catch (Exception e) {
                log.warn("计划入队异常, {}", e.getMessage());
                // 这里一般是数据库失败
            }
        }
    }
}
